package com.lianjia.sh.se.kafka.client;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * Created by yangsp1 on 16/7/27.
 */
public abstract class AbstractKafkaSender{
    Logger logger = LoggerFactory.getLogger(AbstractKafkaSender.class);

    @Autowired
    KafkaProducerFactory kafkaProducerFactory;

    //生产者配置
    private Properties configs = new Properties();

    //主题
    private String topic;

    /**
     * 模拟同步：发送消息(立即刷新，阻塞等待返回结果）
     * @param data 数据
     */
    public void send(String data) throws ExecutionException, InterruptedException {
        ProducerRecord record = new ProducerRecord<>(topic, data);
        Producer producer = kafkaProducerFactory.createProducer(this,configs);
        Future<RecordMetadata> future = producer.send(record);
        producer.flush();//立即发送
        RecordMetadata recordMetadata = future.get();
        logger.info("recordMetadata.offset =" + recordMetadata.offset());
    }

    /**
     * 异步：选缓冲，等待一定条件后刷新缓冲
     * @param data 数据
     */
    public void sendAsyn(String  data){
        ProducerRecord record = new ProducerRecord<>(topic, data);
        Producer producer =  kafkaProducerFactory.createProducer(this,configs);
        producer.send(record);//等待特定时间后发送
    }

    public Properties getConfigs() {
        return configs;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public KafkaProducerFactory getKafkaProducerFactory() {
        return kafkaProducerFactory;
    }

    public void setKafkaProducerFactory(KafkaProducerFactory kafkaProducerFactory) {
        this.kafkaProducerFactory = kafkaProducerFactory;
    }
}
